#pragma once 

/* 实现消息存储子服务部分的封装 */

#include <brpc/server.h>
#include <butil/logging.h>
#include <odb/database.hxx>
#include <odb/mysql/database.hxx>
#include <brpc/channel.h>
#include <unordered_map>

#include "brpc.hpp"     //Rpc调用模块的封装
#include "etcd.hpp"     //服务注册模块封装
#include "logger.hpp"   //日志模块封装
#include "data_es.hpp"  //ES数据管理客户端封装
#include "mysql_message.hpp"   //MySQL数据管理客户端封装
#include "utils.hpp"    //工具类接口
#include "icsearch.hpp" //ES的二次接口封装
#include "rabbitmq.hpp" //RabbitMQ的封装

#include "file.pb.h"    //protobuf框架代码
#include "user.pb.h"    //protobuf框架代码
#include "base.pb.h"    //protobuf框架代码
#include "message.pb.h"    //protobuf框架代码
#include "message.hxx"     //ODB框架代码
#include "message-odb.hxx" //ODB框架代码


namespace ChatSystem
{

    // 1. 创建rpc服务子类继承pb(protobuf)中的MsgStorageService服务类，并实现内部的业务接口逻辑；
    class MsgStorageServiceImpl : public ChatSystem::MsgStorageService
    {
    public:
        MsgStorageServiceImpl(const std::shared_ptr<elasticlient::Client>& es_client,
            const std::shared_ptr<odb::core::database>& mysql_db,
            const ServiceManager::ptr& channel_manager,
            const std::string& file_service_name, 
            const std::string& user_service_name)
            :_es_message(std::make_shared<ESMessage>(es_client))
            ,_mysql_message(std::make_shared<MessageTable>(mysql_db))
            ,_channel_manager(channel_manager)
            ,_file_service_name(file_service_name)
            ,_user_service_name(user_service_name)
        {
            //创建message索引库
            _es_message->createIndex();
        }

        ~MsgStorageServiceImpl(){}

        
        //获取指定时间段的历史消息
        void GetHistoryMsg(google::protobuf::RpcController* controller,
            const ::ChatSystem::GetHistoryMsgReq* request,
            ChatSystem::GetHistoryMsgRsp* response,
            google::protobuf::Closure* done) override
        {
            LOG_DEBUG("收到获取指定时间段的历史消息请求！");
            //使用ClosureGurd类管理done对象，防止出现未调用Run方法；
            brpc::ClosureGuard rpc_gurd(done);
            //定义一个错误处理函数，当出错的时候被调用
            auto err_response = [this, response](const std::string& errmsg)
            {
                response->set_success(false);
                response->set_errmsg(errmsg);
                return;
            };

            std::string request_id = request->request_id();
            response->set_request_id(request_id);

            //1. 从请求中，获取聊天会话ID，要获取的消息的起始时间与结束时间。
            std::string chat_session_id = request->chat_session_id();
            boost::posix_time::ptime start_time(boost::posix_time::from_time_t(request->start_time()));
            boost::posix_time::ptime end_time(boost::posix_time::from_time_t(request->over_time()));
            //判断时间的合法性
            if(start_time.is_not_a_date_time() || end_time.is_not_a_date_time() || start_time > end_time)
            {
                LOG_DEBUG("{} - 指定的时间区间不合法 {} - {} !", request_id, boost::posix_time::to_simple_string(start_time), 
                    boost::posix_time::to_simple_string(end_time));
                return err_response("指定的时间区间不合法！");
            }
            //2. 访问数据库，从数据库中按时间进行范围查询，
            auto msg_lists = _mysql_message->range(chat_session_id, start_time, end_time);
            //获取消息简略信息（消息 ID，会话 ID，消息类型，产生时间，发送者用户 ID，文本消息内容，文件消息元信息）
            std::unordered_set<std::string> file_id_lists; 
            for(const auto& msg : msg_lists)
            {
                if(msg.file_id().empty())
                {
                    continue;
                }
                file_id_lists.insert(msg.file_id());
            }
            std::unordered_map<std::string, std::string> file_data_lists;
            bool ret = _GetFile(request_id, file_id_lists, file_data_lists);
            if(ret == false)
            {
                LOG_DEBUG("{} - 从文件子服务批量文件数据下载失败！", request_id);
                return err_response("从文件子服务批量文件数据下载失败！");
            }
            //3. 循环构造完整消息（从用户子服务获取消息的发送者用户信息，从文件子服务获取文件/语音/图片数据）
            //注意这里要使用std::unordered_set 进行去重，因为从MySQL数据库中查到的消息可能是一个user_id对应多条消息，那么会导致一个 user_id 
            //被重复传入多次，但是从数据库中查找到用户只有一个，就会导致从数据库查找的用户信息数量不一致导致报错；
            std::unordered_set<std::string> user_id_lists;
            for(const auto& msg : msg_lists)
            {
                user_id_lists.insert(msg.user_id());
            }
            std::unordered_map<std::string, UserInfo> user_lists;
            ret = _GetUser(request_id, user_id_lists, user_lists);
            if(ret == false)
            {
                LOG_DEBUG("{} - 从用户管理子服务获取多个用户信息失败！", request_id);
                return err_response("从用户管理子服务获取多个用户信息失败！");
            }
            //4. 组织响应返回给网关服务器。
            for(const auto& msg : msg_lists)
            {
                //add_msg_list函数会给我们new一个MessageInfo对象，将对象的地址返回出来，我们直接对指针指向的内容修改就可以了
                auto message_info = response->add_msg_list();   
                message_info->set_message_id(msg.message_id());
                message_info->set_chat_session_id(msg.chat_session_id());
                message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));
                message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);

                auto message_content =  message_info->mutable_message();

                std::cout << message_info->sender().nickname() << std::endl;
                std::cout << message_info->sender().avatar() << std::endl;

                switch(msg.message_type())
                {
                    case MessageType::STRING:
                    {
                        //文本类型消息
                        message_content->set_message_type(MessageType::STRING);
                        message_content->mutable_string_message()->set_content(msg.content());
                    }
                    break;
                    //图片消息
                    case MessageType::IMAGE:
                    {
                        message_content->set_message_type(MessageType::IMAGE);
                        message_content->mutable_image_message()->set_file_id(msg.file_id());
                        message_content->mutable_image_message()->set_image_contents(file_data_lists[msg.file_id()]);
                    }
                    break;
                    case MessageType::FILE:
                    {
                        //文件消息
                        message_content->set_message_type(MessageType::FILE);
                        message_content->mutable_file_message()->set_file_id(msg.file_id());
                        message_content->mutable_file_message()->set_file_size(msg.file_size());
                        message_content->mutable_file_message()->set_file_name(msg.file_name());
                        message_content->mutable_file_message()->set_file_contents(file_data_lists[msg.file_id()]);
                    }
                    break;
                    case MessageType::SPEECH:
                    {
                        //语音消息
                        message_content->set_message_type(MessageType::SPEECH);
                        message_content->mutable_speech_message()->set_file_id(msg.file_id());
                        message_content->mutable_speech_message()->set_speech_contents(file_data_lists[msg.file_id()]);
                    }
                    break;

                    default:
                        LOG_ERROR("消息类型错误!");
                    return err_response("消息类型错误!"); 
                }
            }
            response->set_success(true);
        }

        //获取最近的N条消息
        void GetRecentMsg(google::protobuf::RpcController* controller,
        const ::ChatSystem::GetRecentMsgReq* request,
        ChatSystem::GetRecentMsgRsp* response,
        google::protobuf::Closure* done) override
        {

            LOG_DEBUG("收到获取最近的N条消息请求!");
            //使用ClosureGurd类管理done对象，防止出现未调用Run方法；
            brpc::ClosureGuard rpc_gurd(done);
            //定义一个错误处理函数，当出错的时候被调用
            auto err_response = [this, response](const std::string& errmsg)
            {
                response->set_success(false);
                response->set_errmsg(errmsg);
                return;
            };

            std::string request_id = request->request_id();
            response->set_request_id(request_id);

            //1. 从请求中，获取聊天会话ID，要获取的消息数量
            std::string chat_session_id = request->chat_session_id();
            int msg_count = request->msg_count();

            //2. 访问数据库，从数据库中按时间排序，
            auto msg_lists = _mysql_message->recent(chat_session_id, msg_count);
            //获取指定数量的消息简略信息(消息ID，会话ID，消息类型，产生时间，发送者用户ID，文本消息内容，文件消息元信息)
            std::unordered_set<std::string> file_id_lists;
            for(const auto& msg : msg_lists)
            {
                if(msg.file_id().empty())
                {
                    continue;
                }
                file_id_lists.insert(msg.file_id());
            }
            std::unordered_map<std::string, std::string> file_data_lists;
            bool ret = _GetFile(request_id, file_id_lists, file_data_lists);
            if(ret == false)
            {
                LOG_DEBUG("{} - 从文件子服务批量文件数据下载失败！", request_id);
                return err_response("从文件子服务批量文件数据下载失败！");
            }
            //3. 循环构造完整消息（从用户子服务获取消息的发送者用户信息，从文件子服务获取文件/语音/图片数据）
            std::unordered_set<std::string> user_id_lists;
            for(const auto& msg : msg_lists)
            {
                user_id_lists.insert(msg.user_id());
            }
            std::unordered_map<std::string, ChatSystem::UserInfo> user_lists;
            ret = _GetUser(request_id, user_id_lists, user_lists);
            if(ret == false)
            {
                LOG_DEBUG("{} - 从用户管理子服务获取多个用户信息失败！", request_id);
                return err_response("从用户管理子服务获取多个用户信息失败！");
            }
            //4. 组织响应返回给网关服务器
            for(const auto& msg : msg_lists)
            {
                //add_msg_list函数会给我们new一个MessageInfo对象，将对象的地址返回出来，我们直接对指针指向的内容修改就可以了
                auto message_info = response->add_msg_list();   
                message_info->set_message_id(msg.message_id());
                message_info->set_chat_session_id(msg.chat_session_id());
                message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));
                message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);


                auto message_content =  message_info->mutable_message();
                switch(msg.message_type())
                {
                    case MessageType::STRING:
                    {
                        //文本类型消息
                        message_content->set_message_type(MessageType::STRING);
                        message_content->mutable_string_message()->set_content(msg.content());
                    }
                    break;
                    //图片消息
                    case MessageType::IMAGE:
                    {
                        message_content->set_message_type(MessageType::IMAGE);
                        message_content->mutable_image_message()->set_file_id(msg.file_id());
                        message_content->mutable_image_message()->set_image_contents(file_data_lists[msg.file_id()]);
                    }
                    break;
                    case MessageType::FILE:
                    {
                        //文件消息
                        message_content->set_message_type(MessageType::FILE);
                        message_content->mutable_file_message()->set_file_id(msg.file_id());
                        message_content->mutable_file_message()->set_file_size(msg.file_size());
                        message_content->mutable_file_message()->set_file_name(msg.file_name());
                        message_content->mutable_file_message()->set_file_contents(file_data_lists[msg.file_id()]);
                    }
                    break;
                    case MessageType::SPEECH:
                    {
                        //语音消息
                        message_content->set_message_type(MessageType::SPEECH);
                        message_content->mutable_speech_message()->set_file_id(msg.file_id());
                        message_content->mutable_speech_message()->set_speech_contents(file_data_lists[msg.file_id()]);
                    }
                    break;

                    default:
                        LOG_ERROR("消息类型错误!");
                    return err_response("消息类型错误!");
                }
            }
            response->set_success(true);
        }

        //搜索指定会话中包含指定关键字的消息
        void MsgSearch(google::protobuf::RpcController* controller,
        const ChatSystem::MsgSearchReq* request,
        ChatSystem::MsgSearchRsp* response,
        google::protobuf::Closure* done) override
        {
            LOG_DEBUG("收到搜索指定会话中包含指定关键字的消息请求！");
            //使用ClosureGurd类管理done对象，防止出现未调用Run方法；
            brpc::ClosureGuard rpc_gurd(done);
            //定义一个错误处理函数，当出错的时候被调用
            auto err_response = [this, response](const std::string& errmsg)
            {
                response->set_success(false);
                response->set_errmsg(errmsg);
                return;
            };

            std::string request_id = request->request_id();
            response->set_request_id(request_id);

            //1. 从请求中，获取聊天会话ID，搜索关键字
            std::string chat_session_id = request->chat_session_id();
            std::string search_key = request->search_key();

            //2. 基于封装的 ES 客户端，访问 ES 服务器进行文本消息搜索（以消息内容进行搜索，以会话 ID 进行过滤），
            //从 ES 服务器获取到消息简息（消息 ID，会话 ID，文本消息内容）
            auto msg_lists = _es_message->search(search_key, chat_session_id);
            //3. 循环从数据库根据消息 ID 获取消息简息（消息 ID，消息类型，会话 ID，发送者ID，产生时间，文本消息内容，文件消息元数据）。
            //在ES索引库中已经存储了消息元信息，不需要从MySQL数据库中查询了，因为先从ES中查询到消息ID后再拿着消息ID到MySQL中查询消息
            // 元信息的方式效率太低下了；

            //4. 循环从用户子服务获取所有消息的发送者用户信息，构造完整消息。
            std::unordered_set<std::string> user_id_lists;
            for (const auto &msg : msg_lists) 
            {
                user_id_lists.insert(msg.user_id());
            }
            std::unordered_map<std::string, ChatSystem::UserInfo> user_lists;
            bool ret = _GetUser(request_id, user_id_lists, user_lists);
            if(ret == false)
            {
                LOG_DEBUG("{} - 从用户管理子服务获取多个用户信息失败！", request_id);
                return err_response("从用户管理子服务获取多个用户信息失败！");
            }
            //5. 组织响应返回给网关服务器
            for(const auto& msg : msg_lists)
            {
                //add_msg_list函数会给我们new一个MessageInfo对象，将对象的地址返回出来，我们直接对指针指向的内容修改就可以了
                auto message_info = response->add_msg_list();   
                message_info->set_message_id(msg.message_id());
                message_info->set_chat_session_id(msg.chat_session_id());
                message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));
                message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);

                auto message_content =  message_info->mutable_message();
                //文本类型消息---只有文本类型的消息
                message_content->set_message_type(MessageType::STRING);
                message_content->mutable_string_message()->set_content(msg.content());
            }  
            response->set_success(true);
        }
    
    public:
        //消息订阅之后执行的回调函数
        void OnMessage(const char* body, size_t sz)
        {
            LOG_DEBUG("在消息队列中收到新消息，进行存储处理!");
            //1. 取出序列化的消息内容，进行反序列化
            ChatSystem::MessageInfo message;
            bool ret = message.ParseFromArray(body, sz);
            if(ret == false)
            {
                LOG_ERROR("对消费的消息进行反序列化失败！");
                return;
            }
            //2. 根据不同的消息类型进行不同的处理
            std::string file_id, file_name, content;
            int64_t file_size;
            switch (message.message().message_type())
            {
                case MessageType::STRING:
                {
                    //2.1 如果是一个文本类型消息，取元信息存储到ES中
                    content = message.message().string_message().content();
                    ret = _es_message->appendData(message.sender().user_id(),
                        message.message_id(), 
                        message.timestamp(), 
                        message.chat_session_id(), 
                        content);
                    if(ret == false)
                    {
                        LOG_ERROR("文本消息向ES索引库进行存储时失败!");
                        return;
                    }
                }
                break;
                //2.2 如果是一个图片/语音/文件消息，则取出数据存储到文件子服务中，并获取对应的文件ID
                case MessageType::IMAGE:
                {
                    const auto& msg = message.message().image_message();
                    ret = _PutFile("", msg.image_contents().size(), msg.image_contents(), file_id);
                    if(ret == false)
                    {
                        LOG_ERROR("图片消息向文件存储子服务上传时失败!");
                        return;
                    }
                }
                break;
                case MessageType::FILE:
                {
                    const auto& msg = message.message().file_message();
                    file_name = msg.file_name();
                    file_size = msg.file_size();
                    ret = _PutFile(file_name, file_size, msg.file_contents(), file_id);
                    if(ret == false)
                    {
                        LOG_ERROR("文件消息向文件存储子服务上传时失败!");
                        return;
                    }
                }
                break;
                case MessageType::SPEECH:
                {
                    const auto& msg = message.message().speech_message();
                    ret = _PutFile("", msg.speech_contents().size(), msg.speech_contents(), file_id);
                    if(ret == false)
                    {
                        LOG_ERROR("语音消息向文件存储子服务上传时失败!");
                        return;
                    }
                }
                break;

                default:
                    LOG_ERROR("消息类型错误!");
                    return;
            }

            //3. 提取消息的元信息，存储到MySQL数据库中
            ChatSystem::Message message_meta_info(message.message_id(), 
                message.chat_session_id(), 
                message.sender().user_id(), 
                message.message().message_type(), 
                boost::posix_time::from_time_t(message.timestamp()));
            message_meta_info.content(content);
            message_meta_info.file_id(file_id);
            message_meta_info.file_name(file_name);
            message_meta_info.file_size(file_size);
            ret = _mysql_message->insert(message_meta_info);
            if(ret == false)
            {
                LOG_ERROR("向数据库插入新消息失败!");
                return;
            }
        }
    private:
        bool _GetUser(const std::string& request_id, 
            const std::unordered_set<std::string>& user_id_lists, 
            std::unordered_map<std::string, UserInfo>& user_lists)
        {
            auto channel = _channel_manager->choose(_user_service_name);
            if(!channel)
            {
                LOG_ERROR("未找到用户管理子服务节点 - {}!", _user_service_name);
                return false;
            }
            ChatSystem::UserService_Stub stub(channel.get());
            ChatSystem::GetMultiUserInfoReq req;
            req.set_request_id(request_id);
            for(const auto& id : user_id_lists)
            {
                req.add_users_id(id);
            }

            brpc::Controller cntl;
            ChatSystem::GetMultiUserInfoRsp rsp;
            stub.GetMultiUserInfo(&cntl, &req, &rsp, nullptr);
            if(cntl.Failed() || rsp.success() == false)
            {
                LOG_ERROR("{} - 用户管理子服务 {} Rpc调用失败, {}!", request_id, _user_service_name, cntl.ErrorText());
                return false;
            }
            
            const auto& user_info_map = rsp.users_info();

            for(auto it = user_info_map.begin(); it != user_info_map.end(); ++it)
            {
                user_lists.insert(std::make_pair(it->first, it->second));     
            }
            return true;
        }

        //实现向文件存储子服务服务器批量获取文件数据
        bool _GetFile(const std::string& request_id, 
            const std::unordered_set<std::string>& file_id_lists, 
            std::unordered_map<std::string, std::string>& file_data_lists)
        {
            auto channel = _channel_manager->choose(_file_service_name);
            if(!channel)
            {
                LOG_ERROR("未找到文件管理子服务节点 - {}!", _file_service_name);
                return false;
            }
            ChatSystem::FileService_Stub stub(channel.get());
            ChatSystem::GetMultiFileReq req;
            req.set_request_id(request_id);
            for(const auto& id : file_id_lists)
            {
                req.add_file_id_list(id);
            }

            brpc::Controller cntl;
            ChatSystem::GetMultiFileRsp rsp;
            stub.GetMultiFile(&cntl, &req, &rsp, nullptr);
            if(cntl.Failed() || rsp.success() == false)
            {
                LOG_ERROR("{} - 文件子服务 {} Rpc调用失败, {}!", request_id, _file_service_name, cntl.ErrorText());
                return false;
            }
            
            const auto& file_data_map = rsp.file_data();
            for(auto it = file_data_map.begin(); it != file_data_map.end(); ++it)
            {
                file_data_lists.insert(std::make_pair(it->first, it->second.file_content()));
            }
            return true;
        }

        //实现向文件存储子服务服务器存储文件数据并获取对应的文件ID
        bool _PutFile(const std::string& file_name, const int64_t file_size, const std::string& body, std::string& file_id)
        {
            //实现文件数据的上传
            auto channel = _channel_manager->choose(_file_service_name);
            if(!channel)
            {
                LOG_ERROR("未找到文件管理子服务节点 - {}!", _file_service_name);
                return false;
            }
            ChatSystem::FileService_Stub stub(channel.get());
            ChatSystem::PutSingleFileReq req;
            req.set_request_id(ChatSystem::Util::uuid());
            req.mutable_file_data()->set_file_name(file_name);
            req.mutable_file_data()->set_file_size(file_size);
            req.mutable_file_data()->set_file_content(body);

            brpc::Controller cntl;
            ChatSystem::PutSingleFileRsp rsp;
            stub.PutSingleFile(&cntl, &req, &rsp, nullptr);
            if(cntl.Failed() || rsp.success() == false)
            {
                LOG_ERROR("{} - 文件子服务 {} Rpc调用失败, {}!", req.request_id(), _file_service_name, cntl.ErrorText());
                return false;
            }
            file_id = rsp.file_info().file_id();
            return true;
        }
        
    private:
        //这些是由内部构造的
        ChatSystem::MessageTable::ptr _mysql_message;     //MySQL的message表， 一定是先实例化odb操作句柄，在实例化这个，注意这个顺序；
        std::shared_ptr<ESMessage> _es_message;     //ES的消息索引

        //这些是Rpc调用客户端相关对象
        std::string _file_service_name; //文件子服务名称
        std::string _user_service_name; //用户管理子服务名称
        ServiceManager::ptr _channel_manager;    //Rpc调用中信道管理对象
        ChatSystem::Discovery::ptr _service_discoverer; //服务发现对象

    };


    //消息存储子服务器搭建的流程类
    class MessageServer
    {
    public:
        using ptr = std::shared_ptr<MessageServer>;

        MessageServer(const Discovery::ptr& discovery_client, 
            const Registry::ptr& registry_client, 
            const std::shared_ptr<elasticlient::Client>& es_client,
            const std::shared_ptr<odb::core::database>& mysql_client,
            const MQClient::ptr& mq_client,
            const std::shared_ptr<brpc::Server>& rpc_server)
        :_discovery_client(discovery_client)
        , _registry_client(registry_client)
        ,_es_client(es_client)
        ,_mysql_client(mysql_client)
        ,_mq_client(mq_client)
        , _rpc_server(rpc_server)
        {}
        ~MessageServer(){}

        //搭建RPC服务器，并启动服务器
        /* make相关接口的调用必须在start接口调用之前，
        因为start涉及到了启动服务器，而启动服务器的前提是客户端对象和服务注册客户端对象已经实例化好了; 
        为了防止使用者在start接口调用之前未调用 make相关接口而导致
        服务器无法启动的问题，我们可以用建造者模式，将构造过程进行控制，在代码中规定调用顺序，去倒逼使用者;
        使用建造者模式的原因还是由于构造userServer对象时参数太多，为了简化userServer对象的构造 */
        void start()
        {
            // 阻塞等待服务端运行结束
            _rpc_server->RunUntilAskedToQuit();
        }
    private:
        Registry::ptr _registry_client;  //注册中心客户端
        Discovery::ptr _discovery_client;    //服务发现客户端
        std::shared_ptr<elasticlient::Client> _es_client;    //ES客户端
        std::shared_ptr<odb::core::database> _mysql_client;    //MySQL的ODB操作句柄
        MQClient::ptr _mq_client;   //RabbitMQ客户端
        std::shared_ptr<brpc::Server> _rpc_server;   //brpc服务器
    };

    //建造者模式---用来构造MessageServer对象，将 MessageServer 对象的真正构造过程给封装起来，通过封装的类去构造 MessageServer 对象
    class MessageServerBuilder
    {
    public:
        //构造es客户端对象
        void make_es_object(const std::vector<std::string>& host_list)
        {
            _es_client = EsClientFactory::create(host_list);
        }

        //构造mysql客户端对象
        void make_mysql_object(const std::string& user,
            const std::string& password,
            const std::string& host,
            const std::string& db,
            const std::string& cset,
            int port,
            int connection_pool_count)
        {
           _mysql_client = ODBFactory::create(user, password, host, db, cset, port, connection_pool_count);
        }

        //构造服务发现客户端对象和信道管理对象
        void make_discovery_object(const std::string& reg_host,
            const std::string& base_service_name,
            const std::string& file_service_name, 
            const std::string& user_service_name)
        {
            _file_service_name = file_service_name;
            _user_service_name = user_service_name;
            _channel_manager = std::make_shared<ServiceManager>();
            _channel_manager->declared(_file_service_name); //将文件管理子服务设置为关心的服务
            _channel_manager->declared(_user_service_name); //将用户管理子服务设置为关心的服务
            LOG_DEBUG("设置文件子服务为需要添加管理的子服务：{}", _file_service_name);
            LOG_DEBUG("设置用户子服务为需要添加管理的子服务：{}", _user_service_name);
            auto put_cb = std::bind(&ChatSystem::ServiceManager::onServiceOnline, 
                _channel_manager.get(), 
                std::placeholders::_1, 
                std::placeholders::_2);
            auto del_cb = std::bind(&ChatSystem::ServiceManager::onServiceOffline, 
                _channel_manager.get(), 
                std::placeholders::_1,
                std::placeholders::_2);
            //2. 构造服务发现对象 
            _discovery_client = std::make_shared<ChatSystem::Discovery>(reg_host, base_service_name, put_cb, del_cb);
        }

        //构造RabbitMQ客户端对象 --- 要进行消息的订阅一定是MySQL操作句柄、ES索引库创建好了之后才进行的，所以消息的订阅应该在消息存储子服务服务器启动的时候
        void make_mq_object(const std::string& user, 
        const std::string& password, 
        const std::string& host, 
        const std::string& exchange_name,   //交换机名称
        const std::string& queue_name,      //消息队列名称
        const std::string& routing_key)      
        {
            _queue_name = queue_name;
            _exchange_name = exchange_name;
           _mq_client = std::make_shared<ChatSystem::MQClient>(user, password, host);
           _mq_client->declareComponents(exchange_name, queue_name, routing_key);

        }

        //用于构建RPC服务器
        void make_rpc_server(uint16_t port, 
            uint32_t timeout,      //连接空闲超时时间---超时后连接关闭
            uint8_t num_threads)    //rpc服务器的线程数量---IO线程数量；
        {
            _rpc_server = std::make_shared<brpc::Server>();
            // _es_client 的实例化必须在 user_service之前进行
            if(!_es_client)
            {
                LOG_ERROR("还未初始化ES搜索模块！");
                abort();
            }
            if(!_mysql_client)
            {
                LOG_ERROR("还未初始化MySQL数据库模块！");
                abort();
            }
            if(!_mq_client)
            {
                LOG_ERROR("还未初始化消息队列客户端模块！");
                abort();
            }
            if(!_channel_manager)
            {
                LOG_ERROR("还未初始化信道管理模块！");
                abort();
            }
            if(!_discovery_client)
            {
                LOG_ERROR("还未初始化服务发现模块！");
                abort();
            }
            // 向rpc服务器类中添加rpc子服务对象---告诉服务器收到什么请求用哪个接口处理；
            /*注意：如果在message_service对象是栈上创建的，AddService的第一个参数就是message_service对象的地址，当
            make_rpc_server函数调用结束后message_service这个对象是被销毁了的，而在 _rpc_server中会使用到 message_service 对象
            此时就会出现段错误，所以我们这里需要在堆上创建对象，这样make_rpc_server函数调用结束后再使用message_service对象也不会错。*/
            MsgStorageServiceImpl* message_service = new MsgStorageServiceImpl(_es_client, 
                _mysql_client, 
                _channel_manager, 
                _file_service_name, 
                _user_service_name);
            /* SERVER_DOESNT_OWN_SERVICE 表示添加服务失败时，服务器也不会删除服务器对象； SERVER_OWNS_SERVICE 表示添加服务失败时，
            服务器将负责删除服务对象，当我们在堆上创建user_service对象时需要使用SERVER_OWNS_SERVICE，就不会造成内存泄漏。 */
            int ret = _rpc_server->AddService(message_service, brpc::ServiceOwnership::SERVER_OWNS_SERVICE); 
            if(ret == -1)
            {
                LOG_ERROR("添加Rpc服务失败！ ");
                abort();
            }
            brpc::ServerOptions options;
            options.idle_timeout_sec - timeout;
            options.num_threads = num_threads;
            ret = _rpc_server->Start(port, &options);
            if(ret == -1)
            {
                LOG_ERROR("服务器启动失败！");
                abort();
            }

            //进行消息队列中消息的订阅 --- 消息的订阅应该在消息存储子服务服务器启动的时候
            auto callback = std::bind(&MsgStorageServiceImpl::OnMessage, message_service, 
                std::placeholders::_1, std::placeholders::_2);
            _mq_client->consume(_queue_name, callback);
        }

         //用于构造服务注册客户端对象
        void make_registry_object(const std::string& reg_host,  //注册中心的地址
            const std::string& service_name,    //服务名称
            const std::string& access_host)     //提供服务器的主机的地址信息
        {
            if(!_rpc_server)
            {
                LOG_ERROR("还未初始化RPC服务器！");
                abort();
            }
            _registry_client = std::make_shared<Registry>(reg_host);
            _registry_client->registry(service_name, access_host);
        }

        //构造一个 MessageServer对象
        MessageServer::ptr build()
        {
            if(!_es_client)
            {
                LOG_ERROR("还未初始化ES搜索模块！");
                abort();
            }
            if(!_mysql_client)
            {
                LOG_ERROR("还未初始化MySQL数据库模块！");
                abort();
            }
            if(!_mq_client)
            {
                LOG_ERROR("还未初始化消息队列客户端模块！");
                abort();
            }
            if(!_channel_manager)
            {
                LOG_ERROR("还未初始化信道管理模块！");
                abort();
            }
            if(!_discovery_client)
            {
                LOG_ERROR("还未初始化服务发现模块！");
                abort();
            }
            if(!_registry_client)
            {
                LOG_ERROR("还未初始化服务注册模块");
                abort();
            }
            if(!_rpc_server)
            {
                LOG_ERROR("还未初始化RPC服务器模块");
                abort();
            }

            MessageServer::ptr server = std::make_shared<MessageServer>(_discovery_client, 
                _registry_client, 
                _es_client,
                _mysql_client,
                _mq_client,
                _rpc_server);

            return server;
        }
    private:
        std::string _file_service_name;  //文件管理子服务名称
        std::string _user_service_name; //用户管理子服务名称

        std::shared_ptr<elasticlient::Client> _es_client;   //ES客户端
        std::shared_ptr<odb::core::database> _mysql_client; //MySQL的ODB操作句柄

        ServiceManager::ptr _channel_manager;   //信道管理对象
        Registry::ptr _registry_client;  //注册中心客户端
        Discovery::ptr _discovery_client;    //服务发现客户端对象

        MQClient::ptr _mq_client;   //消息队列客户端
        std::string _exchange_name;     //交换机名称
        std::string _queue_name;    //消息队列的名称

        std::shared_ptr<brpc::Server> _rpc_server;   //brpc服务器

    };

}